/*
Copyright 2019 The Knative Authors
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
	http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package channel

import (
	"context"
	"errors"
	"fmt"
	"testing"

	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"

	eventingv1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
	"knative.dev/eventing/pkg/apis/feature"

	v1 "knative.dev/eventing/pkg/apis/messaging/v1"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	clientgotesting "k8s.io/client-go/testing"
	"k8s.io/utils/pointer"
	fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
	"knative.dev/pkg/apis"
	duckv1 "knative.dev/pkg/apis/duck/v1"
	v1addr "knative.dev/pkg/client/injection/ducks/duck/v1/addressable"
	"knative.dev/pkg/configmap"
	"knative.dev/pkg/controller"
	fakedynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
	logtesting "knative.dev/pkg/logging/testing"
	"knative.dev/pkg/network"
	. "knative.dev/pkg/reconciler/testing"
	"knative.dev/pkg/tracker"

	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
	"knative.dev/eventing/pkg/client/injection/ducks/duck/v1/channelable"
	channelreconciler "knative.dev/eventing/pkg/client/injection/reconciler/messaging/v1/channel"
	"knative.dev/eventing/pkg/duck"
	. "knative.dev/eventing/pkg/reconciler/testing/v1"
)

const (
	testNS      = "test-namespace"
	channelName = "test-channel"

	readyEventPolicyName   = "test-event-policy-ready"
	unreadyEventPolicyName = "test-event-policy-unready"
)

var (
	testKey = fmt.Sprintf("%s/%s", testNS, channelName)

	deliverySpec = &eventingduckv1.DeliverySpec{
		Retry: pointer.Int32(10),
	}

	backingChannelAddressable = duckv1.Addressable{
		Name: pointer.String("http"),
		URL:  apis.HTTP(network.GetServiceHostname("foo", "bar")),
	}

	channelV1GVK = metav1.GroupVersionKind{
		Group:   "messaging.knative.dev",
		Version: "v1",
		Kind:    "Channel",
	}

	imcV1GVK = metav1.GroupVersionKind{
		Group:   "messaging.knative.dev",
		Version: "v1",
		Kind:    "InMemoryChannel",
	}
)

func TestReconcile(t *testing.T) {

	table := TableTest{{
		Name: "bad workqueue key",
		// Make sure Reconcile handles bad keys.
		Key: "too/many/parts",
	}, {
		Name: "key not found",
		// Make sure Reconcile handles good keys that don't exist.
		Key: "foo/not-found",
	}, {
		Name: "Channel not found",
		Key:  testKey,
	}, {
		Name: "Channel is being deleted",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithChannelDeleted),
		},
	}, {
		Name: "Backing channel tracking error",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRDBadGvk()),
				WithInitChannelConditions),
		},
		WantEvents: []string{
			Eventf(corev1.EventTypeWarning, "InternalError", "unable to track changes to the backing Channel: failed to track"),
		},
		WantErr: true,
	}, {
		Name: "Backing Channel.Create error",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions),
		},
		WantCreates: []runtime.Object{
			createChannelCRD(testNS, channelName, false),
		},
		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
			Object: NewChannel(channelName, testNS,
				WithInitChannelConditions,
				WithChannelTemplate(channelCRD()),
				WithBackingChannelFailed("ChannelFailure", "inducing failure for create inmemorychannels")),
		}},
		WithReactors: []clientgotesting.ReactionFunc{
			InduceFailure("create", "inmemorychannels"),
		},
		WantEvents: []string{
			Eventf(corev1.EventTypeWarning, "InternalError", "problem reconciling the backing channel: %v", "inducing failure for create inmemorychannels"),
		},
		WantErr: true,
	}, {
		Name: "Successful reconciliation",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions),
			NewInMemoryChannel(channelName, testNS,
				WithInitInMemoryChannelConditions,
				WithInMemoryChannelDeploymentReady(),
				WithInMemoryChannelServiceReady(),
				WithInMemoryChannelEndpointsReady(),
				WithInMemoryChannelChannelServiceReady(),
				WithInMemoryChannelAddress(backingChannelAddressable),
				WithInMemoryChannelDLSUnknown(),
				WithInMemoryChannelEventPoliciesReady()),
		},
		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
			Object: NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithBackingChannelReady,
				WithChannelDLSUnknown(),
				WithChannelAddress(&backingChannelAddressable),
				WithChannelEventPoliciesReadyBecauseOIDCDisabled()),
		}},
		Ctx: feature.ToContext(context.Background(), feature.Flags{
			feature.OIDCAuthentication: feature.Disabled,
		}),
	}, {
		Name: "Already reconciled",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithBackingChannelReady,
				WithChannelAddress(&backingChannelAddressable),
				WithChannelDLSUnknown(),
				WithChannelEventPoliciesReadyBecauseOIDCDisabled()),
			NewInMemoryChannel(channelName, testNS,
				WithInitInMemoryChannelConditions,
				WithInMemoryChannelDeploymentReady(),
				WithInMemoryChannelServiceReady(),
				WithInMemoryChannelEndpointsReady(),
				WithInMemoryChannelChannelServiceReady(),
				WithInMemoryChannelSubscribers(subscribers()),
				WithInMemoryChannelAddress(backingChannelAddressable),
				WithInMemoryChannelDLSUnknown(),
				WithInMemoryChannelEventPoliciesReady()),
		},
		Ctx: feature.ToContext(context.Background(), feature.Flags{
			feature.OIDCAuthentication: feature.Disabled,
		}),
	}, {
		Name: "Backing channel created",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithBackingChannelReady,
				WithChannelAddress(&backingChannelAddressable)),
		},
		WantCreates: []runtime.Object{
			createChannel(testNS, channelName, false),
		},
		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
			Object: NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithChannelNoAddress(),
				WithChannelDLSUnknown(),
				WithBackingChannelUnknown("BackingChannelNotConfigured", "BackingChannel has not yet been reconciled."),
				WithChannelEventPoliciesReadyBecauseOIDCDisabled()),
		}},
		Ctx: feature.ToContext(context.Background(), feature.Flags{
			feature.OIDCAuthentication: feature.Disabled,
		}),
	}, {
		Name: "Backing channel created with delivery",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithBackingChannelReady,
				WithChannelDelivery(deliverySpec),
				WithChannelAddress(&backingChannelAddressable)),
		},
		WantCreates: []runtime.Object{
			&unstructured.Unstructured{
				Object: map[string]interface{}{
					"apiVersion": "messaging.knative.dev/v1",
					"kind":       "InMemoryChannel",
					"metadata": map[string]interface{}{
						"namespace": testNS,
						"name":      channelName,
						"ownerReferences": []interface{}{
							map[string]interface{}{
								"apiVersion":         "messaging.knative.dev/v1",
								"blockOwnerDeletion": true,
								"controller":         true,
								"kind":               "Channel",
								"name":               channelName,
								"uid":                "",
							},
						},
					},
					"spec": map[string]interface{}{
						"delivery": map[string]interface{}{
							"retry": int64(10),
						},
					},
				},
			},
		},
		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
			Object: NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithChannelNoAddress(),
				WithChannelDelivery(deliverySpec),
				WithChannelDLSUnknown(),
				WithBackingChannelUnknown("BackingChannelNotConfigured", "BackingChannel has not yet been reconciled."),
				WithChannelEventPoliciesReadyBecauseOIDCDisabled()),
		}},
		Ctx: feature.ToContext(context.Background(), feature.Flags{
			feature.OIDCAuthentication: feature.Disabled,
		}),
	}, {
		Name: "Generation Bump",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithBackingChannelReady,
				WithChannelAddress(&backingChannelAddressable),
				WithChannelGeneration(42)),
			NewInMemoryChannel(channelName, testNS,
				WithInitInMemoryChannelConditions,
				WithInMemoryChannelDeploymentReady(),
				WithInMemoryChannelServiceReady(),
				WithInMemoryChannelEndpointsReady(),
				WithInMemoryChannelChannelServiceReady(),
				WithInMemoryChannelSubscribers(subscribers()),
				WithInMemoryChannelAddress(backingChannelAddressable),
				WithInMemoryChannelDLSUnknown(),
				WithInMemoryChannelEventPoliciesReady()),
		},
		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
			Object: NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithChannelDLSUnknown(),
				WithBackingChannelReady,
				WithChannelAddress(&backingChannelAddressable),
				WithChannelGeneration(42),
				// Updates
				WithChannelObservedGeneration(42),
				WithChannelEventPoliciesReadyBecauseOIDCDisabled()),
		}},
		Ctx: feature.ToContext(context.Background(), feature.Flags{
			feature.OIDCAuthentication: feature.Disabled,
		}),
	}, {
		Name: "Updating subscribers statuses",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithBackingChannelReady,
				WithChannelAddress(&backingChannelAddressable),
				WithChannelEventPoliciesReady()),
			NewInMemoryChannel(channelName, testNS,
				WithInitInMemoryChannelConditions,
				WithInMemoryChannelDuckAnnotationV1Beta1,
				WithInMemoryChannelDeploymentReady(),
				WithInMemoryChannelServiceReady(),
				WithInMemoryChannelEndpointsReady(),
				WithInMemoryChannelChannelServiceReady(),
				WithInMemoryChannelAddress(backingChannelAddressable),
				WithInMemoryChannelSubscribers(subscribers()),
				WithInMemoryChannelStatusSubscribers(subscriberStatuses()),
				WithInMemoryChannelDLSUnknown(),
				WithInMemoryChannelEventPoliciesReady()),
		},
		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
			Object: NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithBackingChannelReady,
				WithChannelAddress(&backingChannelAddressable),
				WithChannelSubscriberStatuses(subscriberStatuses()),
				WithChannelDLSUnknown(),
				WithChannelEventPoliciesReadyBecauseOIDCDisabled()),
		}},
		Ctx: feature.ToContext(context.Background(), feature.Flags{
			feature.OIDCAuthentication: feature.Disabled,
		}),
	}, {
		Name: "Should provision applying EventPolicies",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions),
			NewInMemoryChannel(channelName, testNS,
				WithInitInMemoryChannelConditions,
				WithInMemoryChannelDeploymentReady(),
				WithInMemoryChannelServiceReady(),
				WithInMemoryChannelEndpointsReady(),
				WithInMemoryChannelChannelServiceReady(),
				WithInMemoryChannelAddress(backingChannelAddressable),
				WithInMemoryChannelDLSUnknown(),
				WithInMemoryChannelEventPoliciesReady()),
			NewEventPolicy(readyEventPolicyName, testNS,
				WithReadyEventPolicyCondition,
				WithEventPolicyToRef(channelV1GVK, channelName),
			),
			NewEventPolicy(fmt.Sprintf("%s-%s", readyEventPolicyName, channelName), testNS,
				WithEventPolicyToRef(imcV1GVK, channelName),
				WithEventPolicyOwnerReferences([]metav1.OwnerReference{
					{
						APIVersion: v1.SchemeGroupVersion.String(),
						Kind:       "InMemoryChannel",
						Name:       channelName,
					}, {
						APIVersion: eventingv1alpha1.SchemeGroupVersion.String(),
						Kind:       "EventPolicy",
						Name:       readyEventPolicyName,
					},
				}...),
				WithEventPolicyLabels(map[string]string{
					"messaging.knative.dev/channel-group":   v1.SchemeGroupVersion.Group,
					"messaging.knative.dev/channel-version": v1.SchemeGroupVersion.Version,
					"messaging.knative.dev/channel-kind":    "InMemoryChannel",
					"messaging.knative.dev/channel-name":    channelName,
				}),
			),
		},
		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
			Object: NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithBackingChannelReady,
				WithChannelDLSUnknown(),
				WithChannelAddress(&backingChannelAddressable),
				WithChannelEventPoliciesReady(),
				WithChannelEventPoliciesListed(readyEventPolicyName)),
		}},
	}, {
		Name: "Should mark as NotReady on unready EventPolicies",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions),
			NewInMemoryChannel(channelName, testNS,
				WithInitInMemoryChannelConditions,
				WithInMemoryChannelDeploymentReady(),
				WithInMemoryChannelServiceReady(),
				WithInMemoryChannelEndpointsReady(),
				WithInMemoryChannelChannelServiceReady(),
				WithInMemoryChannelAddress(backingChannelAddressable),
				WithInMemoryChannelDLSUnknown(),
				WithInMemoryChannelEventPoliciesReady()),
			NewEventPolicy(unreadyEventPolicyName, testNS,
				WithUnreadyEventPolicyCondition("", ""),
				WithEventPolicyToRef(channelV1GVK, channelName),
			),
			NewEventPolicy(fmt.Sprintf("%s-%s", unreadyEventPolicyName, channelName), testNS,
				WithEventPolicyToRef(imcV1GVK, channelName),
				WithEventPolicyOwnerReferences([]metav1.OwnerReference{
					{
						APIVersion: v1.SchemeGroupVersion.String(),
						Kind:       "InMemoryChannel",
						Name:       channelName,
					}, {
						APIVersion: eventingv1alpha1.SchemeGroupVersion.String(),
						Kind:       "EventPolicy",
						Name:       unreadyEventPolicyName,
					},
				}...),
				WithEventPolicyLabels(map[string]string{
					"messaging.knative.dev/channel-group":   v1.SchemeGroupVersion.Group,
					"messaging.knative.dev/channel-version": v1.SchemeGroupVersion.Version,
					"messaging.knative.dev/channel-kind":    "InMemoryChannel",
					"messaging.knative.dev/channel-name":    channelName,
				}),
			),
		},
		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
			Object: NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithBackingChannelReady,
				WithChannelDLSUnknown(),
				WithChannelAddress(&backingChannelAddressable),
				WithChannelEventPoliciesNotReady("EventPoliciesNotReady", fmt.Sprintf("event policies %s are not ready", unreadyEventPolicyName))),
		}},
	}, {
		Name: "should list only Ready EventPolicies",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions),
			NewInMemoryChannel(channelName, testNS,
				WithInitInMemoryChannelConditions,
				WithInMemoryChannelDeploymentReady(),
				WithInMemoryChannelServiceReady(),
				WithInMemoryChannelEndpointsReady(),
				WithInMemoryChannelChannelServiceReady(),
				WithInMemoryChannelAddress(backingChannelAddressable),
				WithInMemoryChannelDLSUnknown(),
				WithInMemoryChannelEventPoliciesReady()),
			NewEventPolicy(readyEventPolicyName, testNS,
				WithReadyEventPolicyCondition,
				WithEventPolicyToRef(channelV1GVK, channelName),
			),
			NewEventPolicy(unreadyEventPolicyName, testNS,
				WithUnreadyEventPolicyCondition("", ""),
				WithEventPolicyToRef(channelV1GVK, channelName),
			),
			NewEventPolicy(fmt.Sprintf("%s-%s", readyEventPolicyName, channelName), testNS,
				WithEventPolicyToRef(imcV1GVK, channelName),
				WithEventPolicyOwnerReferences([]metav1.OwnerReference{
					{
						APIVersion: v1.SchemeGroupVersion.String(),
						Kind:       "InMemoryChannel",
						Name:       channelName,
					}, {
						APIVersion: eventingv1alpha1.SchemeGroupVersion.String(),
						Kind:       "EventPolicy",
						Name:       readyEventPolicyName,
					},
				}...),
				WithEventPolicyLabels(map[string]string{
					"messaging.knative.dev/channel-group":   v1.SchemeGroupVersion.Group,
					"messaging.knative.dev/channel-version": v1.SchemeGroupVersion.Version,
					"messaging.knative.dev/channel-kind":    "InMemoryChannel",
					"messaging.knative.dev/channel-name":    channelName,
				}),
			),
			NewEventPolicy(fmt.Sprintf("%s-%s", unreadyEventPolicyName, channelName), testNS,
				WithEventPolicyToRef(imcV1GVK, channelName),
				WithEventPolicyOwnerReferences([]metav1.OwnerReference{
					{
						APIVersion: v1.SchemeGroupVersion.String(),
						Kind:       "InMemoryChannel",
						Name:       channelName,
					}, {
						APIVersion: eventingv1alpha1.SchemeGroupVersion.String(),
						Kind:       "EventPolicy",
						Name:       unreadyEventPolicyName,
					},
				}...),
				WithEventPolicyLabels(map[string]string{
					"messaging.knative.dev/channel-group":   v1.SchemeGroupVersion.Group,
					"messaging.knative.dev/channel-version": v1.SchemeGroupVersion.Version,
					"messaging.knative.dev/channel-kind":    "InMemoryChannel",
					"messaging.knative.dev/channel-name":    channelName,
				}),
			),
		},
		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
			Object: NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithBackingChannelReady,
				WithChannelDLSUnknown(),
				WithChannelAddress(&backingChannelAddressable),
				WithChannelEventPoliciesNotReady("EventPoliciesNotReady", fmt.Sprintf("event policies %s are not ready", unreadyEventPolicyName)),
				WithChannelEventPoliciesListed(readyEventPolicyName)),
		}},
	}, {
		Name: "should create EventPolicies for backing channel",
		Key:  testKey,
		Objects: []runtime.Object{
			NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithChannelEventPoliciesReady(),
				WithChannelEventPoliciesListed(readyEventPolicyName)),
			NewInMemoryChannel(channelName, testNS,
				WithInitInMemoryChannelConditions,
				WithInMemoryChannelDeploymentReady(),
				WithInMemoryChannelServiceReady(),
				WithInMemoryChannelEndpointsReady(),
				WithInMemoryChannelChannelServiceReady(),
				WithInMemoryChannelAddress(backingChannelAddressable),
				WithInMemoryChannelDLSUnknown(),
				WithInMemoryChannelEventPoliciesReady()),
			NewEventPolicy(readyEventPolicyName, testNS,
				WithReadyEventPolicyCondition,
				WithEventPolicyToRef(channelV1GVK, channelName),
				WithEventPolicyFilter(eventingv1.SubscriptionsAPIFilter{
					CESQL: "true",
				}),
			),
		},
		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
			Object: NewChannel(channelName, testNS,
				WithChannelTemplate(channelCRD()),
				WithInitChannelConditions,
				WithBackingChannelObjRef(backingChannelObjRef()),
				WithBackingChannelReady,
				WithChannelDLSUnknown(),
				WithChannelAddress(&backingChannelAddressable),
				WithChannelEventPoliciesReady(),
				WithChannelEventPoliciesListed(readyEventPolicyName)),
		}},
		WantCreates: []runtime.Object{
			NewEventPolicy(fmt.Sprintf("%s-%s", readyEventPolicyName, channelName), testNS,
				WithEventPolicyToRef(imcV1GVK, channelName),
				WithEventPolicyOwnerReferences([]metav1.OwnerReference{
					{
						APIVersion: v1.SchemeGroupVersion.String(),
						Kind:       "InMemoryChannel",
						Name:       channelName,
					}, {
						APIVersion: eventingv1alpha1.SchemeGroupVersion.String(),
						Kind:       "EventPolicy",
						Name:       readyEventPolicyName,
					},
				}...),
				WithEventPolicyLabels(map[string]string{
					"messaging.knative.dev/channel-group":   v1.SchemeGroupVersion.Group,
					"messaging.knative.dev/channel-version": v1.SchemeGroupVersion.Version,
					"messaging.knative.dev/channel-kind":    "InMemoryChannel",
					"messaging.knative.dev/channel-name":    channelName,
				}),
				WithEventPolicyFilter(eventingv1.SubscriptionsAPIFilter{
					CESQL: "true",
				}),
			),
		},
	}}

	logger := logtesting.TestLogger(t)
	table.Test(t, MakeFactory(func(ctx context.Context, listers *Listers, cmw configmap.Watcher) controller.Reconciler {
		ctx = channelable.WithDuck(ctx)
		ctx = v1addr.WithDuck(ctx)
		r := &Reconciler{
			dynamicClientSet:   fakedynamicclient.Get(ctx),
			channelLister:      listers.GetMessagingChannelLister(),
			channelableTracker: &fakeListableTracker{duck.NewListableTrackerFromTracker(ctx, channelable.Get, tracker.New(func(types.NamespacedName) {}, 0))},
			eventPolicyLister:  listers.GetEventPolicyLister(),
			eventingClientSet:  fakeeventingclient.Get(ctx),
		}
		return channelreconciler.NewReconciler(ctx, logger,
			fakeeventingclient.Get(ctx), listers.GetMessagingChannelLister(),
			controller.GetEventRecorder(ctx), r)
	},
		false,
		logger,
	))
}

type fakeListableTracker struct {
	duck.ListableTracker
}

func (f *fakeListableTracker) TrackInNamespaceKReference(ctx context.Context, obj metav1.Object) duck.TrackKReference {
	return func(ref duckv1.KReference) error {
		if ref.Kind == channelCRDBadGvk().Kind {
			return errors.New("failed to track")
		}
		return f.ListableTracker.TrackInNamespaceKReference(ctx, obj)(ref)
	}
}

func channelCRD() metav1.TypeMeta {
	return metav1.TypeMeta{
		APIVersion: "messaging.knative.dev/v1",
		Kind:       "InMemoryChannel",
	}
}

func channelCRDBadGvk() metav1.TypeMeta {
	return metav1.TypeMeta{
		APIVersion: "broken.testing.dev",
		Kind:       "FailsTrack",
	}
}

func subscribers() []eventingduckv1.SubscriberSpec {
	return []eventingduckv1.SubscriberSpec{{
		UID:           "2f9b5e8e-deb6-11e8-9f32-f2801f1b9fd1",
		Generation:    1,
		SubscriberURI: apis.HTTP("call1"),
		ReplyURI:      apis.HTTP("sink2"),
	}, {
		UID:           "34c5aec8-deb6-11e8-9f32-f2801f1b9fd1",
		Generation:    2,
		SubscriberURI: apis.HTTP("call2"),
		ReplyURI:      apis.HTTP("sink2"),
	}}
}

func subscriberStatuses() []eventingduckv1.SubscriberStatus {
	return []eventingduckv1.SubscriberStatus{{
		UID:                "2f9b5e8e-deb6-11e8-9f32-f2801f1b9fd1",
		ObservedGeneration: 1,
		Ready:              "True",
	}, {
		UID:                "34c5aec8-deb6-11e8-9f32-f2801f1b9fd1",
		ObservedGeneration: 2,
		Ready:              "True",
	}}
}

func createChannelCRD(namespace, name string, ready bool) *unstructured.Unstructured {
	u := &unstructured.Unstructured{
		Object: map[string]interface{}{
			"apiVersion": "messaging.knative.dev/v1",
			"kind":       "InMemoryChannel",
			"metadata": map[string]interface{}{
				"namespace": namespace,
				"name":      name,
				"ownerReferences": []interface{}{
					map[string]interface{}{
						"apiVersion":         "messaging.knative.dev/v1",
						"blockOwnerDeletion": true,
						"controller":         true,
						"kind":               "Channel",
						"name":               name,
						"uid":                "",
					},
				},
			},
		},
	}
	if ready {
		u.Object["status"] = map[string]interface{}{
			"address": map[string]interface{}{
				"url": backingChannelAddressable.URL.String(),
			},
		}
	}
	return u
}

func backingChannelObjRef() *duckv1.KReference {
	return &duckv1.KReference{
		APIVersion: "messaging.knative.dev/v1",
		Kind:       "InMemoryChannel",
		Namespace:  testNS,
		Name:       channelName,
	}
}

func createChannel(namespace, name string, ready bool) *unstructured.Unstructured {
	var hostname string
	var url string
	if ready {
		return &unstructured.Unstructured{
			Object: map[string]interface{}{
				"apiVersion": "messaging.knative.dev/v1",
				"kind":       "InMemoryChannel",
				"metadata": map[string]interface{}{
					"namespace": namespace,
					"name":      name,
					"ownerReferences": []interface{}{
						map[string]interface{}{
							"apiVersion":         "messaging.knative.dev/v1",
							"blockOwnerDeletion": true,
							"controller":         true,
							"kind":               "Channel",
							"name":               name,
							"uid":                "",
						},
					},
				},
				"status": map[string]interface{}{
					"address": map[string]interface{}{
						"hostname": hostname,
						"url":      url,
					},
				},
			},
		}
	}

	return &unstructured.Unstructured{
		Object: map[string]interface{}{
			"apiVersion": "messaging.knative.dev/v1",
			"kind":       "InMemoryChannel",
			"metadata": map[string]interface{}{
				"namespace": namespace,
				"name":      name,
				"ownerReferences": []interface{}{
					map[string]interface{}{
						"apiVersion":         "messaging.knative.dev/v1",
						"blockOwnerDeletion": true,
						"controller":         true,
						"kind":               "Channel",
						"name":               name,
						"uid":                "",
					},
				},
			},
		},
	}
}
